package com.mutouren.common.mq.kafka;

import java.io.Closeable;
import java.io.IOException;
import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class Producer<K, V> implements Closeable{
	private KafkaProducer<K, V> producer;	
	private Properties props;
	
	@SuppressWarnings("unchecked")
	public Producer(String bootstrapServers) {
		this(bootstrapServers, (Class<K>)Integer.class, (Class<V>)String.class);
	}
	
	@SuppressWarnings("unchecked")
	public Producer(String bootstrapServers, Class<V> valueType) {
		this(bootstrapServers, (Class<K>)Integer.class, valueType);
	}	
	
	public Producer(String bootstrapServers, Class<K> keyType, Class<V> valueType) {
		this(getProperties(bootstrapServers, keyType, valueType));
	}	
	
	public Producer(Properties props) {
		this.props = props;
		producer = new KafkaProducer<K, V>(this.props);
	}
	
	public static String getSerializer(Class<?> clazz) {
		String result;
		
		if (clazz== Integer.class) {
			result = "org.apache.kafka.common.serialization.IntegerSerializer";
		} else if (clazz == Long.class) {
			result = "org.apache.kafka.common.serialization.LongSerializer";
		} else if (clazz == byte[].class) {
			result = "org.apache.kafka.common.serialization.ByteArraySerializer";	
		} else if (clazz == String.class) {
			result = "org.apache.kafka.common.serialization.StringSerializer";
		} else {
			throw new IllegalArgumentException( "producer serializer don't know the type: " + clazz);
		}		
		return result;
	}
	
	private static Properties getProperties(String bootstrapServers, Class<?> keyType, Class<?> valueType) {
		Properties props = new Properties();
		props.put("bootstrap.servers", bootstrapServers);
        props.put("key.serializer", getSerializer(keyType));
        props.put("value.serializer", getSerializer(valueType));
//        props.put("acks", 1);
//        props.put("retries", 0);
//        props.put("compression.type", "none");
        
		return props;
	}
	
	public void send(String topic, V value) {
		this.send(topic, null, value);
	}	
	public void send(String topic, K key, V value) {
		this.send(topic, null, key, value);
	}		
	public void send(String topic, Integer partition, K key, V value) {
		producer.send(new ProducerRecord<K, V>(topic, partition, key, value));
	}
	
	public void send(String topic, V value, Callback callback) {
		this.send(topic, null, value, callback);
	}	
	public void send(String topic, K key, V value, Callback callback) {
		this.send(topic, null, key, value, callback);
	}		
	public void send(String topic, Integer partition, K key, V value, Callback callback) {
		producer.send(new ProducerRecord<K, V>(topic, partition, key, value), callback);
	}	
	
	public void flush() {
		this.producer.flush();
	}

	@Override
	public void close() throws IOException {
		this.producer.close();		
	}
	
}
